package com.yh.config;

import com.yh.config.properties.ConsumerProperties;
import com.yh.service.impl.CustomMsgServiceImpl;
import lombok.AllArgsConstructor;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @author yh
 */
@Configuration
@AllArgsConstructor
public class ConsumerMqConfig {

    public static final Logger LOGGER = LoggerFactory.getLogger(ConsumerMqConfig.class);

    private final ConsumerProperties consumerProperties;
    private final CustomMsgServiceImpl customMsgService;

    @Bean
    public DefaultMQPushConsumer defaultConsumer() throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerProperties.getGroupName());
        consumer.setNamesrvAddr(consumerProperties.getServerAddress());
        consumer.setConsumeThreadMax(consumerProperties.getConsumeThreadMax());
        consumer.setConsumeThreadMin(consumerProperties.getConsumeThreadMin());
        consumer.registerMessageListener(customMsgService);
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        consumer.subscribe(consumerProperties.getSubscribe(),"*");
        consumer.start();
        return consumer;
    }
}
